package com.atbeijing.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark23_Oper_Req {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("TransformOperator")
        val sc = new SparkContext(conf)

        val startTime = System.currentTimeMillis()
        // TODO 需求 ： 统计出每一个省份每个广告被点击数量排行的Top3
        // 先聚合再分组

        // 1. 读取日志文件，获取原始数据
        //    line => 时间戳，省份，城市，用户，广告
        val rdd : RDD[String] = sc.textFile("data/agent.log")

        // 2. 将数据转换结构
        //    line => (省份-广告, 1)
        val mapRDD : RDD[(String, Int)] = rdd.map(
            line => {
                val datas = line.split(" ")
                (datas(1) + "-" + datas(4), 1)
            }
        )

        // 3. 将转换结构后的数据进行分组聚合
        //     (省份-广告, 1), (省份-广告, 1) =>  (省份-广告, sum)
        val reduceRDD : RDD[(String, Int)] = mapRDD.reduceByKey(_+_)

        // 4. 将聚合的结果进行结构的转换
        //     (省份-广告, sum) =>  (省份, (广告, sum))
        val mapRDD1 = reduceRDD.map {
            case ( key, sum ) => {
                val ks = key.split("-")
                (ks(0), (ks(1), sum))
            }
        }

        // 5. 将转换后的数据根据省份进行分组
        //    (省份, List[(广告1, sum1),(广告2, sum2),(广告3, sum3)])
        val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD1.groupByKey()
//
//        // 6. 将分组后的数据根据点击数量进行排序(降序)，取前3名
        val top3 = groupRDD.mapValues(
            iter => {
                iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
            }
        )

        // 7. 每个省份，广告排行
        //    (河北，List( (A, 100), (B, 50), (C, 20) ))
        //    (河南，List( (C, 100), (A, 50), (B, 20) ))
        top3.collect().foreach(println)

        sc.stop()
        println(System.currentTimeMillis() - startTime)
    }

}
